package com.tn.rtdata.stomp;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.kaazing.gateway.client.protocol.stomp.StompClient;

import com.kaazing.gateway.client.protocol.stomp.StompEvent;
import com.kaazing.gateway.client.protocol.stomp.StompListener;
import com.tn.openrpc.RemoteMethod;
import com.tn.openrpc.TimeoutException;
import com.tn.openrpc.serialization.SerializationException;
import com.tn.openrpc.serialization.Serializer;
import com.tn.openrpc.stomp.StompException;
import com.tn.openrpc.stomp.StompTemplate;
import com.tn.openrpc.util.Factory;
import com.tn.openrpc.util.FactoryException;
import com.tn.rtdata.DataEvent;
import com.tn.rtdata.EventDrivenDataSource;
import com.tn.rtdata.FillException;
import com.tn.rtdata.Session;

/**
 * An implementation of <code>DataSource</code> that receives <code>DataEvent</code>s via STOMP.
 */
public class StompDataSource<T> extends EventDrivenDataSource<T>
{
  private static final Logger LOGGER = LoggerFactory.getLogger(StompDataSource.class);

  private static final String DEFAULT_ACKNOWLEDGEMENT = "auto";
  private static final String DESTINATION_TYPE_POSTFIX = "://";
  private static final String HEADER_ID = "id";

  private FillEventMonitor fillEventMonitor;
  private StompClient stompClient;
  private String stompFillEventDestination;
  private Map<String, String> subscribeHeaders;
  private Serializer<String> serializer;

  /**
   * Creates a new <code>StompDataSource</code>.
   *
   * @param session                    the associated session.
   * @param fillName                   the name of the fill.
   * @param fillParameters             the parameters of the fill.
   * @param fillStartMethod            the <code>RemoteMethod</code> that starts the fill.
   * @param fillStopMethod             the <code>RemoteMethod</code> that stops the fill.
   * @param dataEventStatsUpdateMethod the <code>RemoteMethod</code> that called to update the data-event stats.
   * @param stompClient                the stomp client used to receive fill event messages.
   * @param stompFillEventDestination  the destination used to receive fill event messages.
   * @param serializer                 the serializer used to deserialize the fill events.
   */
  public StompDataSource(
    Session session,
    String fillName,
    List<?> fillParameters,
    RemoteMethod fillStartMethod,
    RemoteMethod fillStopMethod,
    RemoteMethod dataEventStatsUpdateMethod,
    StompClient stompClient,
    String replyToEchoDestination,
    String stompFillEventDestination,
    Factory<String> idFactory,
    Serializer<String> serializer
  )
    throws FactoryException, TimeoutException, StompException
  {
    super(
      session,
      fillName,
      fillParameters,
      fillStartMethod,
      fillStopMethod,
      dataEventStatsUpdateMethod,
      toDestination(stompClient, idFactory, replyToEchoDestination, stompFillEventDestination)
    );
    this.stompClient = stompClient;
    this.stompFillEventDestination = stompFillEventDestination;
    this.serializer = serializer;
    this.fillEventMonitor = new FillEventMonitor();
    this.subscribeHeaders = createSubscriptionHeaders(idFactory);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void startFillEventMonitor() throws FillException
  {
    stompClient.subscribe(stompFillEventDestination, DEFAULT_ACKNOWLEDGEMENT, null, subscribeHeaders);
    stompClient.addStompListener(fillEventMonitor);

    LOGGER.debug("Subscribed for fill events to destination: {}", stompFillEventDestination);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void stopFillEventMonitor() throws FillException
  {
    stompClient.removeStompListener(fillEventMonitor);
    stompClient.unsubscribe(stompFillEventDestination, null, subscribeHeaders);

    LOGGER.debug("Unsubscribed for fill events from destination: {}", stompFillEventDestination);
  }

  /**
   * Creates the subscription headers.
   */
  private Map<String, String> createSubscriptionHeaders(Factory<String> idFactory) throws FactoryException
  {
    Map<String, String> subscribeHeaders = new HashMap<String, String>();

    subscribeHeaders.put(HEADER_ID, idFactory.newInstance());

    return subscribeHeaders;
  }

  /**
   * Converts the <code>stompDestination</code> to an JMS destination name.
   */
  private static String toDestination(
    StompClient stompClient,
    Factory<String> idFactory,
    String replyToEchoDestination,
    String stompFillEventDestination
  )
    throws TimeoutException, StompException
  {
    LOGGER.info("Resolving stomp fill event destination: {}", stompFillEventDestination);

    String destinationName = new StompTemplate(stompClient, idFactory).send(
      replyToEchoDestination,
      stompFillEventDestination,
      "fill.replyToEcho",
      StringUtils.EMPTY
    );

    LOGGER.info("Resolved stomp fill event destination: {} to: {}", stompFillEventDestination, destinationName);

    return destinationName.substring(destinationName.indexOf(DESTINATION_TYPE_POSTFIX) + DESTINATION_TYPE_POSTFIX.length());
  }

  /**
   * An implementation of <code>StompListener</code> that processes messages received and fires <code>DataEvent</code>s in response to them.
   */
  private class FillEventMonitor implements StompListener
  {
    /**
     * {@inheritDoc}
     */
    @Override
    public void onOpen(StompEvent stompEvent)
    {
      //Required be implemented interface.
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onMessage(StompEvent stompEvent)
    {
      try
      {
        Object messageContent = serializer.deserialize(stompEvent.getBodyAsText());
        if (messageContent instanceof Iterable)
        {
          for (Object messageContentItem : (Iterable)messageContent)
          {
            processMessageContent(messageContentItem);
          }
        }
        else if (messageContent instanceof DataEvent)
        {
          processMessageContent(messageContent);
        }
      }
      catch (SerializationException e)
      {
        LOGGER.error("An error occurred de-serializing the message contents.", e);
      }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onReceipt(StompEvent stompEvent)
    {
      //Required be implemented interface.
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onClose(StompEvent stompEvent)
    {
      //Required be implemented interface.
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onError(StompEvent stompEvent)
    {
      LOGGER.error("ERROR: {}", stompEvent.getHeaders());
    }
  }
}
